package com.quanyan.stadium.component.mq;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.google.gson.Gson;
import com.quanyan.common.utils.ObjectUtil;
import com.quanyan.mq.enums.MqConsumerStatus;
import com.quanyan.mq.mq.consumer.AbstractMqConsumer;
import com.quanyan.mq.mq.consumer.MqTopic;
import com.quanyan.stadium.constants.GameConstants;
import com.quanyan.stadium.entity.vo.StadiumNotifyMsg;
import com.quanyan.stadium.service.commonComponent.DataReductionService;
import com.quanyan.stadium.service.commonComponent.GameMQLog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * Created by xiaoxiao on 16/6/13.
 */
@Component(value = "stadiumSelfConsumer")
public class StadiumSelfConsumer extends AbstractMqConsumer {

    private final static Logger logger = LoggerFactory.getLogger(StadiumSelfConsumer.class);

    @Value("${mq.nameServer.address}")
    private String nameServerAddress;

    @Value("${mq.nameServer.consumerGroup}")
    private String producerGroup;

    @Value("${mq.nameServer.consumerInstanceName}")
    private String instanceName;

    @Autowired
    private DataReductionService dataReductionService;

    @Autowired
    private GameMQLog gameMQLog;

    @PostConstruct
    void run() {
        super.start();
    }

    @Override
    protected MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        /**
         * 实现自己的逻辑业务
         *
         * */
        /**
         *  默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
         *  consumer.setConsumeMessageBatchMaxSize(X);
         */
        MessageExt msg = messages.get(0);
        String msgId = msg.getMsgId();
        String msgKeys = msg.getKeys();
        if (!gameMQLog.isConsumerFinish(msgKeys)) {
            byte[] bodyByte = msg.getBody();
            StadiumNotifyMsg stadiumNotifyMsg = (StadiumNotifyMsg) ObjectUtil.toObject(bodyByte);
            Gson gson = new Gson();
            logger.info("运动场积分消息和用户个人消息消费，msgId：{} ，msgKeys: {}, 消息主题: {}, 消息tag: {}, 消息内容：{} ", msgId, msgKeys, msg.getTopic(), msg.getTags(), gson.toJson(stadiumNotifyMsg));
            try{
                if (GameConstants.STADIUM_TOPIC_ROUND.equals(msg.getTopic()) && GameConstants.STADIUM_TOPIC_ROUND_RECORD.equals(msg.getTags())) {
                    if (GameConstants.STADIUM_TAG_ROUND_RECORD.equals(stadiumNotifyMsg.getMsgType())) {  //赛事结束栏目积分
                        dataReductionService.gameScheduleCategoryScoreRecord(stadiumNotifyMsg);
                    } else if (GameConstants.STADIUM_TAG_SCORE_GRADE.equals(stadiumNotifyMsg.getMsgType())) {  //赛程结束评级积分
                        dataReductionService.gameScheduleScoreRecord(stadiumNotifyMsg);
                    } else if (GameConstants.STADIUM_TAG_USER_SCHEDULE_INIT_SCHEDULE.equals(stadiumNotifyMsg.getMsgType())) {  // 初始化
                        dataReductionService.userGameScheduleInit(stadiumNotifyMsg);
                    } else if (GameConstants.STADIUM_TAG_USER_SCHEDULE_UPDATE_SCHEDULE.equals(stadiumNotifyMsg.getMsgType())) {  //更新赛程
                        dataReductionService.userGameScheduleUpdate(stadiumNotifyMsg);
                    } else if (GameConstants.STADIUM_TAG_USER_SCHEDULE_FINALLY_SCHEDULE.equals(stadiumNotifyMsg.getMsgType())) {  //最终比分
                        dataReductionService.userGameScheduleFinally(stadiumNotifyMsg);
                    } else if (GameConstants.STADIUM_TAG_GAME_EVENT_FINISH_INIT_GAME_SCORE.equals(stadiumNotifyMsg.getMsgType())) {  //赛事结束,开始执行一些操作
                        dataReductionService.userGameEventCollect(stadiumNotifyMsg);
                    } else if (GameConstants.STADIUM_TAG_USER_PK_GAME_COLLECT.equals(stadiumNotifyMsg.getMsgType())) {  //记录用户PK赛事次数
                        dataReductionService.userPKGameEventCollect(stadiumNotifyMsg);
                    } else if (GameConstants.STADIUM_TAG_GAME_RADAR_CHART.equals(stadiumNotifyMsg.getMsgType())) {  //记录雷达图
                        dataReductionService.updateGameRadarChartBatch(stadiumNotifyMsg);
                    } else if (GameConstants.STADIUM_TAG_GAME_SCHEDULE_END_PUSH_MSG.equals(stadiumNotifyMsg.getMsgType())) {   //轮次完结的消息推送和相关MQ
                        dataReductionService.generateCurrentRoundPushMsgAndMQ(stadiumNotifyMsg);
                    }else if (GameConstants.STADIUM_TAG_GAME_EVENT_FINISH_JOIN_INTO_MONTH.equals(stadiumNotifyMsg.getMsgType())) {   //排位赛积分录入到月赛事中
                        dataReductionService.officialGameEventTeamJoinUpdate(stadiumNotifyMsg);
                    }else if (GameConstants.STADIUM_TAG_GAME_EVENT_FINISH_PUSH_MSG.equals(stadiumNotifyMsg.getMsgType())) {   //赛事结束,开始推送相关信息
                        dataReductionService.officialGameEventPushMsg(stadiumNotifyMsg);
                    }else if (GameConstants.STADIUM_TAG_GAME_EVENT_FINISH_MSG_COLLECT.equals(stadiumNotifyMsg.getMsgType())) {   //赛事结束开始数据收集
                        dataReductionService.officialGameEventCollectMsg(stadiumNotifyMsg);
                    }
                } else if (GameConstants.STADIUM_TOPIC_ROUND.equals(msg.getTopic()) && GameConstants.STADIUM_TOPIC_PK_RANK_PAGE.equals(msg.getTags())) {
                    dataReductionService.generateNextPagePKRankScore(stadiumNotifyMsg);             //设置分页
                }
                gameMQLog.updateMqFinishByMsgId(msgKeys); //更新赛事为消费状态  (msgId和msgKeys不一样)
            }catch (Exception e) {
                logger.error("运动场消息,消费过程中出现异常,当前执行的topic:{},当前执行的tag:{},当前消息体的类型:{},错误原因:{}",
                        msg.getTopic(), msg.getTags(), stadiumNotifyMsg.getMsgType(), e);
            }
        }
        return MqConsumerStatus.CONSUMER_SUCCESS;
    }

    @Override
    protected String getConsumerGroup() {
        return this.producerGroup;
    }

    @Override
    protected String getNameServerAddress() {
        return nameServerAddress;
    }

    @Override
    protected String getInstanceName() {
        return this.instanceName;
    }

    @Override
    protected MqTopic[] getMqTopics() {
        MqTopic[] mqTopics = new MqTopic[1];
        mqTopics[0] = new MqTopic(GameConstants.STADIUM_TOPIC_ROUND, "*");
        return mqTopics;
    }
}
